package com.shengzai.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3DataFrameAPI {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .appName("api")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    /**
     * 1.读取数据
     */
    val studentInfo: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("id string,name string, age int ,sex string ,clazz string ")
      .load("data/students.txt")

    val scoreInfo: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("sid string,cid string, score int")
      .load("data/score.txt")


    /**
     * 2、show：打印部分数据
     * show函数里面有一个truncate参数 将参数设置为false 就可以看到当前行所有的数据（可能出现不会对齐的现象）
     */

    studentInfo.show(100,false)


    /**
     *3.select:选择数据
     */
    import org.apache.spark.sql.functions._
    import spark.implicits._

    studentInfo.select("id","name","clazz").show()

    //使用列对象方法，可以对列数据进行操作
    studentInfo.select($"name",$"age",$"age"+1 as "res").show()

    /**
     * 4.where 过滤数据
     */

    //1.字符串形式
    studentInfo.where("age=22").show()

    //2.列对象形式
    studentInfo.where($"age"===23).show()

    /**
     * 5.groupBy 聚合
     * 聚合和聚合函数一起用
     */

      //使用agg函数聚合计算班级人数和平均年龄
    studentInfo
      .groupBy("clazz")
      .agg(count($"*") as "num",round(avg($"age"),2) as "avg_age")
      .show()

    //可以使用where 代替having
    studentInfo
      .groupBy("clazz")
      .agg(count($"*") as "num", round(avg($"age"), 2) as "avg_age")
     // .where("num >90 and avg_age >21") 可以通过字符串方法
      .where($"num".>(90)and $"avg_age".>(21))
      .show()

    /**
     * 6.join 表关联
     */
    // 当两个连接字段相同使用下面方法,连接后结果不会包含两个字段
    // studentInfo.join(scoreInfo,"id").show()

    //当两个连接自动不相同使用下面的方法，连接后会包含两个相同的字段
    studentInfo.as("a") .join(scoreInfo.as("b"),$"a.id"===$"b.sid").show()

    /**
     * 7.orderBy: 排序
     */
    studentInfo.groupBy($"clazz" as "clazz")
      .agg(count($"*") as "count" )
      .orderBy($"count".desc)
      .show()

    /**
     * 8.limit
     */
    studentInfo.limit(10).show()

    /**
     * 9.withColumn :在原来的表中添加列，并且可以进行逻辑运算
     */
    studentInfo.withColumn("resAge",$"age"+10).show()

    /**
     * 10.row_number
     */

    //计算每个班级年龄最大的前十个学生

    studentInfo.
      withColumn("pm",row_number() over Window.partitionBy($"clazz").orderBy($"age".desc))
      .where("pm<=10")
      .limit(100)
      .show()

  }

}
